package com.sprucetec.live.mcq.category;

import com.sprucetec.live.mcq.McqBusinessMemoryMap;
import com.sprucetec.live.block.ChatMessageConsumer;
import com.sprucetec.live.common.dto.business.LiveChatDto;
import com.sprucetec.live.service.LiveChatService;
import com.sprucetec.live.utils.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 阻塞队列消费策略 并发太高 线程池（chatCustomerThreadPool） 策略 会将部分消息丢弃
 */
@Slf4j
@Component
public class ConsumerProcess implements LiveChatMegProcessCategory {

    private Lock lock = new ReentrantLock();

    @Autowired
    private ThreadPoolTaskExecutor liveChatThreadPool;

    private static final Integer queueContainer = 200000;

    @Override
    public void processChatMessage(LiveChatDto liveChatDto) {
        BlockingQueue<LiveChatDto> liveChatDtoBlockingQueue = null;
        try {
            lock.lock();
            liveChatDtoBlockingQueue = McqBusinessMemoryMap.queueHashMap.get(liveChatDto.getVideoId());
            if (liveChatDtoBlockingQueue == null) {
                LinkedBlockingQueue<LiveChatDto> queue = new LinkedBlockingQueue<>(queueContainer);
                McqBusinessMemoryMap.queueHashMap.put(liveChatDto.getVideoId(), queue);
                ChatMessageConsumer chatMessageConsumer = (ChatMessageConsumer) SpringContextUtil.getBean("chatMessageConsumer");
                chatMessageConsumer.setQueue(queue);
                liveChatDtoBlockingQueue = queue;
                log.info("MCQ消费端阻塞队列初始化成功");
                chatMessageConsumer.setVideoId(liveChatDto.getVideoId());
                liveChatThreadPool.execute(chatMessageConsumer);
            }
        } catch (Exception ex) {
            log.info("阻塞队列消费失败=> {}", ex);
        } finally {
            lock.unlock();
        }
        if (liveChatDtoBlockingQueue != null) {
            liveChatDtoBlockingQueue.offer(liveChatDto); // 将聊天信息放入阻塞队列
        }

    }

    @Override
    public void close() {

    }
}
